package com.spark.core.transformation

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * cogroup操作将两个RDD按照它们的键（这里是名字）进行联接，将具有相同键的元素组合在一起。
 * resultRDD是一个包含了键和对应值的元组的RDD，值是一个元组，其中包含了两个可迭代的集合，
 * 一个是rdd1中具有相同键的所有性别信息的集合，另一个是rdd2中具有相同键的所有年龄信息的集合。
 * 和fullOuterJoin类似, 只不过cogroup的一个key就是一条记录
 */
object Demo16_cogroup {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
    conf.setMaster("local")
    conf.setAppName("cogroup")
    val sc = new SparkContext(conf)
    sc.setLogLevel("error")

    val rdd1 = sc.parallelize(List[(String, String)](
      ("zhangsan", "female"),
      ("zhangsan", "female1"),
      ("lisi", "male"),
      ("wangwu", "female"),
      ("zhaoliu", "male")), 3)
    val rdd2 = sc.parallelize(List[(String, Int)](("zhangsan", 18),
      ("lisi", 19), ("lisi", 190),
      ("wangwu", 20), ("tianqi", 21)), 4)

    val resultRDD: RDD[(String, (Iterable[String], Iterable[Int]))] = rdd1.cogroup(rdd2)

    resultRDD.foreach(elem => {
      val key: String = elem._1
      val list1: List[String] = elem._2._1.toList
      val list2: List[Int] = elem._2._2.toList
      println("key = " + key + ",iter1 = " + list1 + ",iter2 = " + list2)
    })


  }
}
